package bx.job.actors;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import bx.job.lock.bean.ScheduleLockContextBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;

public class JobScheduleActor extends AbstractBehavior<JobScheduleActor.JobScheduleActorMsg> {

    private static final ScheduleJobMsg SCHEDULE_JOB_MSG = new ScheduleJobMsg();
    private final static Logger log = LoggerFactory.getLogger(JobScheduleActor.class);

    public static final int DEFAULT_SCHEDULE_TIME = 3;

    private ActorRef<JobIssueActor.JobIssueActorMsg> jobIssueActor;

    private ActorRef<JobLogActor.JobIssueActorMsg> jobLogActor;

    private ActorRef<JobTriggerActor.JobTriggerActorMsg> jobTriggerActor;

    private ActorRef<LockManagerActor.LockManagerActorMsg> lockManagerActor;

    public interface JobScheduleActorMsg {
    }

    public static class ScheduleJobMsg implements JobScheduleActorMsg {
    }

    public static record LockSuccessMsg(ScheduleLockContextBean lockContext) implements JobScheduleActorMsg {
    }


    public static class StopScheduleMsg implements JobScheduleActorMsg {
    }


    public static Behavior<JobScheduleActor.JobScheduleActorMsg> create() {
        return Behaviors.setup(context -> Behaviors.withTimers(timer -> {
            timer.startTimerWithFixedDelay(SCHEDULE_JOB_MSG, SCHEDULE_JOB_MSG, Duration.ofSeconds(DEFAULT_SCHEDULE_TIME));
            return new JobScheduleActor(context);
        }));
    }

    private JobScheduleActor(ActorContext<JobScheduleActor.JobScheduleActorMsg> context) {
        super(context);
        this.jobTriggerActor = context.spawn(JobTriggerActor.create(), "bx-job-JobTriggerActor");
        this.jobIssueActor = context.spawn(JobIssueActor.create(), "bx-job-JobIssueActor");
        this.jobLogActor = context.spawn(JobLogActor.create(), "bx-job-JobLogActor");
        this.lockManagerActor = context.spawn(LockManagerActor.create(), "bx-job-LockManagerActor");
    }

    @Override
    public Receive<JobScheduleActorMsg> createReceive() {
        return newReceiveBuilder().onMessage(ScheduleJobMsg.class, scheduleJobMsg -> {
                    ScheduleLockContextBean scheduleLockContext = ScheduleLockContextBean.builder().build();
                    this.lockManagerActor.tell(new LockManagerActor.LockMsg(scheduleLockContext, this.getContext().getSelf()));
                    return this;
                }).onMessage(LockSuccessMsg.class, msg -> {
                    ActorRef<JobProcessManagerActor.JobProcessManagerActorMsg> processManagerActor =
                            this.getContext().spawn(JobProcessManagerActor.create(lockManagerActor, jobLogActor, jobIssueActor),
                                    "bx-job-JobProcessManagerActor");
                    JobTriggerActor.TriggerJobMsg triggerJobMsg = new JobTriggerActor.TriggerJobMsg(processManagerActor);
                    this.jobTriggerActor.tell(triggerJobMsg);
                    return this;
                }).
                onMessage(StopScheduleMsg.class,
                        stopScheduleMsg -> Behaviors.stopped()
                )
                .build();
    }
}
